package com.godo.hadoop.merger;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;

import com.godo.hadoop.config.SpiderConfiguration;


/** 
 * by dyong 2010-6-29
 */
public abstract class BaseMerger {
	protected static final Log log = LogFactory.getLog(BaseMerger.class);
	protected static final String date = new SimpleDateFormat("yyyy-MM-dd").format(new Date()) ;

	protected FileSystem fs ;
	protected Job job ;
	protected Path inputPath ;
	protected ArrayList<Path> inputPaths ;
	protected Path outputPath ;
	
	public abstract void merger() throws IOException, InterruptedException, ClassNotFoundException ;
	public BaseMerger() {
		Configuration conf = SpiderConfiguration.create();
		try {
			fs = FileSystem.get(conf);
			job = new Job(conf,this.getClass().getSimpleName());
		} catch(IOException e) {
			e.printStackTrace();
		}
	}
	protected void deleteDirectory(String file){
		File f = new File(file) ;
		if(f.exists()){
			deleteDirectory(f) ;
		}
	}
	protected void deleteDirectory(File file){
		if(file.isDirectory()){
			File[] files = file.listFiles() ;
			if(files!=null)
			for(File f:files){
				if(f.isDirectory()){
					deleteDirectory(f) ;
				} else {
					f.delete() ;
				}
			}
		}
		if(file.exists()){
			file.delete() ;
		}
//		file.deleteOnExit() ;
	}
	
	protected List<File> getDirFileList(File file){
		ArrayList<File> list = new ArrayList<File>() ;
		if(file.isFile()){
			list.add(file) ;
		} else {
			File[] files = file.listFiles() ;
			for(File f:files){
				list.addAll(getDirFileList(f)) ;
			}
		}
		return list ;
	}
	protected void createFile(File file) throws IOException{
		if(file.exists()){
			return ;
		}
		file.mkdirs() ;
	}
}